package cn.com.sparkle.paxos.net.netlayer.netty;

import java.util.List;


import cn.com.sparkle.paxos.net.frame.FrameBody;
import cn.com.sparkle.paxos.net.frame.FrameHead;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

/**
 * ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 * | 4bytes = 4bit checksumtype + 28bit body length | 8bit head checksum     |body      | ceil(body length / ChecksumUtil.CHECK_CHUNK_SIZE) * 8 or 0 bytes body checksum ,checksum type is defined in head |
 * ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 * @author qinan.qn
 *
 */
public class CheckSumPackageDecoder extends ByteToMessageDecoder {
	
//	private Logger logger = Logger.getLogger(CheckSumPackageDecoder.class);
	
	private final static AttributeKey<FrameHead> frameHeadKey = new AttributeKey<FrameHead>("framehead");
	private byte[] nullBytes = new byte[0];

	public CheckSumPackageDecoder() {
	}

	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		Attribute<FrameHead> attr = ctx.attr(frameHeadKey);
		FrameHead frameHead = ctx.attr(frameHeadKey).get();
		for (;;) {
			if (frameHead != null) {
				if (in.readableBytes() >= frameHead.getBodySerializeSize()) {
					byte[] body = new byte[frameHead.getBodySize()];
					byte[] checksum = new byte[frameHead.getBodyChecksumLength()];
					in.readBytes(body);
					in.readBytes(checksum);
					FrameBody frameBody = new FrameBody(body, checksum, frameHead.getChecksumType());
					out.add(frameBody);
					frameHead = null;
					attr.set(null);
				} else {
					break;
				}
			}
			if (in.readableBytes() >= 4) {
				int readerIndex = in.readerIndex();
				int checksumLength = FrameHead.calcChecksumLength(in.getByte(readerIndex));
				if (in.readableBytes() >= checksumLength + 4) {
					byte[] head = new byte[4];
					byte[] checksum;
					in.readBytes(head);
					if (checksumLength == 0) {
						checksum = nullBytes;
					} else {
						checksum = new byte[checksumLength];
						in.readBytes(checksum);
					}
//					in.readBytes(head);
					frameHead = new FrameHead(head, checksum);
					if (frameHead.isValid()) {
						ctx.attr(frameHeadKey).set(frameHead);
					} else {
						//frame checksum error, close connnection
						ctx.close();
					}
				} else {
					break;
				}
			} else {
				break;
			}
		}
	}

}
